package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.commont.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/4/25 09:36
 */
public class Dwd_03_DwdTradeCartAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdTradeCartAdd().init(
            30003,
            2,
            "Dwd_03_DwdTradeCartAdd"
        );
        
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_03_DwdTradeCartAdd");
        // 2. 过滤加购数据: insert 和 update(sku_num变化并且sku_num 变大)
        Table cartAdd = tEnv.sqlQuery(
            "select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['sku_id'] sku_id, " +
                " if(`type`='insert', " +
                "    cast(`data`['sku_num'] as int), " +
                "    cast(`data`['sku_num'] as int) - cast(`old`['sku_num'] as int) " +
                "    ) sku_num, " +
                " ts " +
                "from ods_db " +
                "where `database`='gmall2023' " +
                "and `table`='cart_info' " +
                "and (`type`='insert' " +
                " or (`type`='update' " +
                "     and `old`['sku_num'] is not null " +
                "     and cast(`data`['sku_num'] as int ) > cast(`old`['sku_num'] as int)" +
                "    )" +
                ")");
        
        // 3. 查询到的结果写入到 kafka 中
        tEnv.executeSql(
            "CREATE TABLE dwd_trade_cart_add ( " +
                "    id string, " +
                "    user_id string, " +
                "    sku_id string, " +
                "    sku_num int, " +
                "    ts bigint)" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_CART_ADD)
        );
    
        cartAdd.executeInsert("dwd_trade_cart_add");
        
    }
}
/*
加购事务事实表:
	来源: ods_db 中的 cart_info

	操作类型: insert 新加购商品
			 update 已加购的商品的数量增加的情况

商品 id       数量
sku_1    	   2       insert   -> 写入到 dwd 层的 kafka
sku_1          5       update   -> 写入到 dwd 层的 kafka  有问题: dws 聚和结果会出问题, 应该写的新增量
sku_1          5-2     update  -> 变化后-变化前  > 0

将来在 dws 做聚合
	统计商品被加购的数量:
		sku_1     7
 */